package com.kancy.stream.rabbit.config;

import com.kancy.stream.rabbit.converter.MessageConverter;
import com.kancy.stream.rabbit.exception.MessageException;
import com.kancy.stream.rabbit.handler.ExceptionHandler;
import com.kancy.stream.rabbit.handler.MessageHandler;
import com.kancy.stream.rabbit.message.ChannelMessage;
import com.kancy.stream.rabbit.message.SimpleChannelMessage;
import com.kancy.stream.rabbit.properties.ChannelProperties;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;

import java.util.Objects;

/**
 * RabbitMessageChannelListener
 *
 * @author kancy
 * @date 2020/3/8 0:38
 */
public class RabbitMessageChannelListener implements MessageListener {

    private final ChannelProperties channelProperties;
    private final DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver;

    public RabbitMessageChannelListener(ChannelProperties channelProperties, DynamicInterfaceBeanResolver dynamicInterfaceBeanResolver) {
        this.channelProperties = channelProperties;
        this.dynamicInterfaceBeanResolver = dynamicInterfaceBeanResolver;
    }

    @Override
    public void onMessage(Message message) {
        try {
            doMessage(message);
        } catch (Exception e) {
            ExceptionHandler exceptionHandler = findExceptionHandler(message);
            if (Objects.nonNull(exceptionHandler)){
                exceptionHandler.onException(message, e);
            }
        }
    }

    private void doMessage(Message message) {
        MessageProperties messageProperties = message.getMessageProperties();

        String messageType = (String) messageProperties.getHeaders().get(ChannelMessage.HEADER_MESSAGE_TYPE);
        String channelKey = (String) messageProperties.getHeaders().get(ChannelMessage.HEADER_CHANNEL_KEY);

        if (Objects.nonNull(messageType)){
            // 消息处理器
            MessageHandler messageHandler = findMessageHandler(messageType);
            // 封装ChannelMessage
            MessageConverter messageConverter = findChannelMessageConverter(messageType);
            ChannelMessage channelMessage = new SimpleChannelMessage(channelKey);
            channelMessage.setPayload(messageConverter.fromMessage(message));
            channelMessage.setHeaders(messageProperties.getHeaders());
            messageHandler.onMessage(channelMessage);
        }else {
            throw new MessageException(
                    "Channel Message property [message-type] could not be found, this is an illegal channel message.");
        }
    }

    private ExceptionHandler findExceptionHandler(Message message) {
        return dynamicInterfaceBeanResolver.getBean(
                channelProperties.getConsumer()
                        .get(message.getMessageProperties().getHeaders().get(ChannelMessage.HEADER_MESSAGE_TYPE))
                        .getExceptionHandler(), ExceptionHandler.class);
    }

    private MessageConverter findChannelMessageConverter(String messageType) {
        return dynamicInterfaceBeanResolver.getBean(
                channelProperties.getConsumer().get(messageType).getMessageConverter(),
        MessageConverter.class);
    }

    private MessageHandler findMessageHandler(String messageType) {
        return dynamicInterfaceBeanResolver.getBean(
                channelProperties.getConsumer().get(messageType).getMessageHandler(), MessageHandler.class);
    }

}
